package org.lql.chaos.kafka.example.design;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

import java.nio.charset.StandardCharsets;

/**
 * @author: liangqinglong
 * @date: 2025-08-10 14:33
 * @description: 统一消息发送接口实现类
 **/
@Service
public class MessageProducerServiceImpl implements MessageProducerService {

	@Autowired
	private KafkaTemplateManager kafkaTemplateManager;

	@Autowired
	private PartitionStrategyManager partitionStrategyManager;

	@Override
	public <K, V> ListenableFuture<SendResult<K, V>> send(String topic, K key, V payload, String serializationType, String partitionStrategy) {
		KafkaTemplate<K, V> kafkaTemplate = kafkaTemplateManager.getKafkaTemplate(serializationType);

		// 构造 ProducerRecord 并添加分区策略到 Header
		ProducerRecord<K, V> record = new ProducerRecord<>(topic, null, key, payload, new RecordHeaders());
		record.headers().add("partition-strategy", partitionStrategy.getBytes(StandardCharsets.UTF_8));

		// 也可以添加更多自定义header，比如业务ID、traceId等

		return kafkaTemplate.send(record);
	}
}
